package com.shujia.streeam

import org.apache.hadoop.conf.Configuration
import org.apache.hadoop.hbase.CellUtil
import org.apache.hadoop.hbase.client.{Get, HConnectionManager, Put}
import org.apache.hadoop.hbase.util.Bytes
import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.sql.SQLContext
import org.apache.spark.streaming.{Durations, StreamingContext}

object Demo4uodataStateByKeyOnHBase {

  def main(args: Array[String]): Unit = {


    /**
      * 使用hbase存储中间状态
      *
      */

    val conf = new SparkConf().setMaster("local[2]").setAppName("Demo1WordCount")
    conf.set("spark.sql.shuffle.partitions", "2")
    val sc = new SparkContext(conf)


    val sqlContext = new SQLContext(sc)

    import sqlContext.implicits._

    /**
      *
      * 创建 streamming上下文对象  需要指定间隔时间   延迟时间
      *
      */
    val ssc = new StreamingContext(sc, Durations.seconds(5))

    /**
      * 读取socket创建DS
      * nc -lk 8888
      *
      */

    val studentDS = ssc.socketTextStream("node1", 8888)


    /**
      * foreachRDD  不是一个算子   但是每隔5秒也会执行一次
      *
      */
    studentDS.foreachRDD(rdd => {

      //统计每个班级的人数
      val kvRDD = rdd.map(line => (line.split(",")(4), 1))
      val countRDD = kvRDD.reduceByKey((x, y) => x + y)


      //将统计结果存到hbase
      countRDD.foreachPartition(iter => {

        //创建hbase连接
        val config = new Configuration()
        config.set("hbase.zookeeper.quorum", "node4:2181,node2:2181,node3:2181")
        val connectionManager = HConnectionManager.createConnection(config)
        //create 'clazz_index' ,'info'
        val table = connectionManager.getTable("clazz_index")
        iter.foreach(kv => {

          var curr = kv._2

          val rowkey = kv._1.getBytes()

          val get = new Get(rowkey)

          //如果这个key在hbase中已经有数据  将数据读取出来和当前batch的结果相加  最后在写回去

          val result = table.get(get)
          val value = Bytes.toString(result.getValue("info".getBytes(), "num".getBytes))

          if (value != null) {
            println("==============" + value)
            val last = value.toInt
            curr = curr + last
          }

          //班级作为rowkey
          val put = new Put(rowkey)

          //将最后的结果写会hbase
          put.add("info".getBytes(), "num".getBytes(), curr.toString.getBytes())
          table.put(put)

          println(kv._1 + "\t" + curr)
        })

        connectionManager.close()

      })


    })

    ssc.start()
    ssc.awaitTermination()
    ssc.stop()

  }
}
